package com.adcc.ags.cgw.managed.downlink;

import com.adcc.ags.cgw.configuration.AMQSFactory;
import com.adcc.ags.cgw.configuration.UserListFactory;
import com.adcc.ags.cgw.entity.mq.MQInfo;
import com.adcc.ags.cgw.entity.mq.MQState;
import com.adcc.ags.cgw.entity.msg.CGWMsg;
import com.adcc.ags.cgw.entity.msg.MsgType;
import com.adcc.ags.cgw.managed.proxy.MsgProxy;
import com.adcc.utility.log.Log;
import com.adcc.utility.mq.configuration.MQConfiguration;
import com.adcc.utility.mq.configuration.MQConfigurationFactory;
import com.adcc.utility.mq.entity.Message;
import com.adcc.utility.mq.transfer.MQConnectionPool;
import com.adcc.utility.mq.transfer.MQConnectionPoolFactory;
import com.adcc.utility.mq.transfer.MQTransfer;
import com.adcc.utility.mq.transfer.ibm.IBMMQConnectionPool;
import com.adcc.utility.mq.transfer.ibm.IBMMQTransfer;
import com.google.common.collect.Lists;
import io.dropwizard.lifecycle.Managed;
import io.dropwizard.setup.Environment;

import java.util.List;

/**
 * 网关下行处理Managed
 */
public class DownlinkForward implements Managed {

    // 下行线程名称
    public static final String CONSTANT_DOWNLINK_THREAD_NAME = "Thread.Downlink";

    // 单例对象
    private static DownlinkForward instance;

    // 下行队列
    private String queueAMQSToGW;

    // 上行队列
    private String queueGWToAMQS;

    // AMQS配置信息
    private List<ConnectionFactory> amqsFactoryList = Lists.newArrayList();

    // AMQS状态
    private MQInfo amqs = new MQInfo();

    // MQTransfer
    private MQTransfer transfer = new IBMMQTransfer();

    // MQConfiguration
    private MQConfiguration configuration;

    // MQConnectionPool
    private MQConnectionPool pool;

    // 下行处理线程
    private DownlinkProcessRunnable dpr;

    /**
     * 构造方法
     * */
    private DownlinkForward() {

    }

    /**
     * 单例方法
     * */
    public static DownlinkForward getInstance() {
        if (instance == null) {
            instance = new DownlinkForward();
        }
        return instance;
    }

    public MQInfo getAMQS(){
        return amqs;
    }

    /**
     * 建立Managed
     * @param environment
     */
    public void build(Environment environment, AMQSFactory amqsFactory) throws Exception {
        if (environment == null) {
            throw new NullPointerException("environment is null");
        }
        if (amqsFactory == null) {
            throw new NullPointerException("downlinkFactory is null");
        }

        // 设置AMQS上下行队列配置
        queueAMQSToGW = amqsFactory.getAmqsFactoryList().get(0).getRecvQueue();
        queueGWToAMQS = amqsFactory.getAmqsFactoryList().get(0).getSendQueue();
        // 设置AMQS集群配置
        for(ConnectionFactory connection : amqsFactory.getAmqsFactoryList()){
            ConnectionFactory af = new ConnectionFactory();
            af.setHost(connection.getHost());
            af.setPort(connection.getPort());
            af.setHost(connection.getHost());
            af.setQueueManager(connection.getQueueManager());
            af.setChannel(connection.getChannel());
            af.setRecvQueue(connection.getRecvQueue());
            af.setSendQueue(connection.getSendQueue());
            amqsFactoryList.add(af);
        }
        // 设置AMQS连接池配置，连接AMQS
        pool = MQConnectionPoolFactory.getInstance().createIBMMQConnectionPool();
        ((IBMMQConnectionPool)pool).setActiveMode(amqsFactory.getActiveMode());
        ((IBMMQConnectionPool)pool).setTimeout(amqsFactory.getTimeout());
        ((IBMMQConnectionPool)pool).setMaxConnections(amqsFactory.getMaxConnections());
        ((IBMMQConnectionPool)pool).setMaxIdelConnections(amqsFactory.getMaxIdelConnections());
        transfer.setConnectionPool(pool);
        connectAMQS();

        environment.lifecycle().manage(this);
    }

    /**
     * 连接AMQS
     */
    private boolean connectAMQS() {
        try {
            for (ConnectionFactory af : amqsFactoryList) {
                configuration = MQConfigurationFactory.getInstance().createIBMMQConfiguration(af.getHost(), af.getPort(), af.getQueueManager(), af.getChannel());
                transfer.setConfiguration(configuration);
                pool.init(configuration);

                if (transfer.isConnected()) {
                    Log.info(DownlinkForward.class.getName(), "amqs connected host:" + af.getHost() + " port:" + af.getPort() + " channel:" + af.getChannel() + " qm:" + af.getQueueManager());
                    amqs.setHost(af.getHost());
                    amqs.setPort(af.getPort());
                    amqs.setChannel(af.getChannel());
                    amqs.setQm(af.getQueueManager());
                    amqs.setState(MQState.CONNECTED);
                    return true;
                }
            }
            return false;
        } catch (Exception ex) {
            Log.error(DownlinkForward.class.getName(), "connectAMQS() error", ex);
            return false;
        }
    }

    /**
     * 发送消息到AQMS
     * @param msg
     * @throws Exception
     */
    public void sendMsgToAMQS(byte[] msg) throws Exception {
        try {
            Message message = new Message(msg);
            transfer.sendQueue(queueGWToAMQS, message);
        } catch (Exception ex) {
            Log.error(DownlinkForward.class.getName(), "sendMsgToAMQS() error", ex);
            throw ex;
        }
    }

    @Override
    public void start() throws Exception {
        try {
            dpr = new DownlinkProcessRunnable();
            new Thread(dpr,CONSTANT_DOWNLINK_THREAD_NAME).start();
        } catch (Exception ex) {
            Log.error(DownlinkForward.class.getName(), "start() error", ex);
        }
    }

    @Override
    public void stop() throws Exception {
        try {
            if (dpr != null) {
                dpr.close();
            }
            amqs.setState(MQState.DISCONNECTED);
            pool.dispose();
        } catch (Exception ex) {
            Log.error(DownlinkForward.class.getName(), "stop() error", ex);
        }
    }

    /**
     * 下行处理线程
     */
    class DownlinkProcessRunnable implements Runnable {

        // 线程启动标识
        private boolean isStarted;

        /**
         * 关闭线程
         */
        public void close() {
            isStarted = false;
        }

        @Override
        public void run() {
            try {
                Log.info(DownlinkProcessRunnable.class.getName(), "DownlinkProcessRunnable start...");
                isStarted = true;
                while (isStarted) {
                    try {
                        if (amqs.getState() == MQState.CONNECTED) {
                            if (UserListFactory.getInstance().getUserList().size() > 0) {

                                // 下行收报
                                Message message = null;
                                try {
                                    message = transfer.receiveQueue(queueAMQSToGW);
                                } catch (Exception ex) {
                                    Log.error(DownlinkProcessRunnable.class.getName(), "receive message from amqs error", ex);
                                    amqs.setState(MQState.CONNECTING);
                                }

                                // 下行报处理
                                if (message != null) {
                                    Log.info(DownlinkProcessRunnable.class.getName(), "receive message from amqs\r\n" + message);
                                    CGWMsg cgwMsg = new CGWMsg(message, MsgType.Downlink, "1");
                                    MsgProxy.getInstance().getMsgRouteActor().tell(cgwMsg,MsgProxy.getInstance().getMsgRouteActor());
                                }
                            }else{
                                Thread.sleep(50);
                            }
                        } else {
                            // 尝试连接AMQS
                            if (!connectAMQS()) {
                                amqs.setState(MQState.CONNECTING);
                            }
                            Thread.sleep(50);
                        }
                    } catch (Exception ex) {
                        Log.error(DownlinkProcessRunnable.class.getName(), "process downlink message error", ex);
                    }
                }
            } catch (Exception ex) {
                Log.error(DownlinkProcessRunnable.class.getName(), "run() error", ex);
            }
        }
    }
}
